package com.tjf.myjetpackdemo.RxObserverModel;

/**
 * @author: tjf
 * @date: 2023-06-07
 * @desc：用来创建被观察者对象
 */
public class ObservableCreate<T> extends Observable<T> {
    ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    //如果订阅功能要实现 在这里面通知 observer消息更新
    @Override
    public void subscribeActual(Observer<T> observer) {
        CreateEmitter createEmitter = new CreateEmitter(observer);
        observer.onSubscribe();//订阅
        //把发射器和被观察者绑定在一起
        source.subscribe(createEmitter);
    }

    //发射器  把发射器和观察者建立联系
    static final class CreateEmitter<T> implements Emitter<T> {
        public CreateEmitter(Observer<T> observer) {
            this.observer = observer;
        }

        Observer<T> observer;

        @Override
        public void onNext(T t) {
            observer.onNext(t);
        }

        @Override
        public void onError(Throwable throwable) {
            observer.onError(throwable);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }

}
